package com.alison.sparkstream.transform

import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.DStream
import org.apache.spark.streaming.{Seconds, StreamingContext}

object E2_transform3_window {

  """
    |有状态转换操作1：滑动窗口转换操作
    |
    |1，基本介绍
    |（1）滑动窗口操作是一种允许我们对一段时间内的数据执行操作的机制，窗口会根据时间滑动，以便持续分析流数据。
    |（2）所有基于窗口的操作都需要两个参数，分别为窗口时长以及滑动步长。
    |窗口时长：计算内容的时间范围
    |滑动步长：隔多久触发一次计算。
    |
    |注意：
    |“窗口时长”和“滑动步长”都必须为采集周期大小的整数倍。
    |在滑动窗口中，数据可能会被多次计算，需要小心处理，避免重复计算。
    |
    |
    |主题： 我们会对连续 10 秒内的数据进行单词频度计算，并且每 5 秒滑动一次。
    |步骤：
    |1. 我们每隔 5 秒，依次输入数据 a、b、c，
    |window(windowLength, slideInterval)
    |
    |
    |2. 计数
    |countByWindow(windowLength, slideInterval)
    |
    |3. 元素计数并返回 DStream[String, count]
    |countByValueAndWindow(windowLength, slideInterval)
    |
    |
    |""".stripMargin


  def main(args: Array[String]): Unit = {
    windower
  }

  // 创建 Spark 运行配置对象
  val sparkConf = new SparkConf().setMaster("local[*]").setAppName("Hello")

  // 初始化 StreamingContext，设置微批处理的时间间隔为 5 秒
  val ssc = new StreamingContext(sparkConf, Seconds(5))

  def windower() = {
    // 设置检查点以保存状态
    ssc.checkpoint("./checkpoint")
    // 通过监控端口创建 DStream，读进来的数据为一行行
    val inputDStream: DStream[String] = ssc.socketTextStream("192.168.56.104", 9999)
    // 对 DStream 进行转换操作，将每一行数据做切分，形成一个个单词
    val wordStreams = inputDStream.flatMap(_.split(" "))
    // 将单词映射成元组（word,1）
    val wordAndOneStreams = wordStreams.map((_, 1))
    // 使用window函数进行滑动窗口操作
    """
      |-------------------1----------------------
      |window(windowLen, slideInterval)
      |""".stripMargin
    //    使用了 window 函数来创建一个滑动窗口，窗口长度为 10 秒，滑动间隔为 5 秒。
    val windowedCounts = wordAndOneStreams.window(Seconds(10), Seconds(5)).reduceByKey(_ + _)
    // 打印结果
    windowedCounts.print

    // 启动 StreamingContext
    ssc.start()
    // 等待应用程序终止（遇错停止，否则不断工作）
    ssc.awaitTermination()
  }

  def countByWindower() = {
    """
      |------------------2---------------------
      |countByWindow(windowLen, slideInterval)
      |
      |""".stripMargin
    ssc.checkpoint("./checkpoint")
    // 通过监控端口创建 DStream，读进来的数据为一行行
    val inputDStream: DStream[String] = ssc.socketTextStream("192.168.56.104", 9999)
    // 对 DStream 进行转换操作，将每一行数据做切分，形成一个个单词
    val wordStreams = inputDStream.flatMap(_.split(" "))
    // 该方法返回一个滑动窗口计数流中的元素个数
    val cntStream = wordStreams.countByWindow(Seconds(10), Seconds(5))
    cntStream.print
    // 启动 StreamingContext
    ssc.start()
    // 等待应用程序终止（遇错停止，否则不断工作）
    ssc.awaitTermination()
  }

  def countByValueAndWindower() = {
    // 通过监控端口创建 DStream，读进来的数据为一行行
    val inputDStream: DStream[String] = ssc.socketTextStream("192.168.56.104", 9999)
    // 对 DStream 进行转换操作，将每一行数据做切分，形成一个个单词
    val wordStreams = inputDStream.flatMap(_.split(" "))
    // 使用window函数进行滑动窗口操作
    val countByValueStream = wordStreams.countByValueAndWindow(Seconds(10), Seconds(5))
    // 打印结果
    countByValueStream.print()
    // 启动 StreamingContext
    ssc.start()
    // 等待应用程序终止（遇错停止，否则不断工作）
    ssc.awaitTermination()
  }
  """
    |------------------3---------------------
    |reduceByWindow(func, windowLength, slideInterval)
    |
    |该方法通过使用自定义的函数 func 来整合滑动窗口中的批次数据，创建一个新的单元素流。这个函数会对窗口内的数据应用 func 函数，以生成一个结果。
    |""".stripMargin

  def reduceByWindower() = {
    // 通过监控端口创建 DStream，读进来的数据为一行行
    val inputDStream: DStream[String] = ssc.socketTextStream("192.168.56.104", 9999)
    // 对 DStream 进行转换操作，将每一行数据做切分，形成一个个单词
    val wordStreams = inputDStream.flatMap(_.split(" "))
    // 对 DStream 进行转换操作，将每一行数据做切分，形成一个个数字
    val numStream = wordStreams.map(_.toInt)
    val reducedStream = numStream.reduceByWindow((a, b) => a + b, Seconds(10), Seconds(5))
    reducedStream.print
    // 启动 StreamingContext
    ssc.start()
    // 等待应用程序终止（遇错停止，否则不断工作）
    ssc.awaitTermination()
  }

  """
    |reduceByKeyAndWindow(func, windowLength, slideInterval, [numTasks])
    |
    |该方法是在 (K, V) 键值对 DStream 上使用的，它会返回一个新的 (K, V) 键值对 DStream，其中的每个窗口都通过使用 reduce 函数来整合每个键的值。
    |func 函数将被应用于窗口内的批次数据。
    |""".stripMargin

  def reduceByKeyAndWindower() = {
    // 通过监控端口创建 DStream，读进来的数据为一行行
    val inputDStream: DStream[String] = ssc.socketTextStream("192.168.56.104", 9999)
    // 对 DStream 进行转换操作，将每一行数据做切分，形成一个个单词
    val wordStreams = inputDStream.flatMap(_.split(" "))
    // 将单词映射成元组（word,1）
    val wordPairsStreams = wordStreams.map((_, 1))
    // 使用window函数进行滑动窗口操作
    val reducedStream = wordPairsStreams.reduceByKeyAndWindow((a, b) => a + b, Seconds(10), Seconds(5))
    // 打印结果
    reducedStream.print()
    // 启动 StreamingContext
    ssc.start()
    // 等待应用程序终止（遇错停止，否则不断工作）
    ssc.awaitTermination()
  }

  """
    |该方法是 reduceByKeyAndWindow 方法的变体，它允许我们通过使用先前窗口的 reduce 值来递增计算当前窗口的 reduce 值。这可以在数据增量计算中很有用。
    |""".stripMargin
  def reduceByKeyAndWindowerInvFunc() = {
    // 通过监控端口创建 DStream，读进来的数据为一行行
    val inputDStream: DStream[String] = ssc.socketTextStream("localhost", 9999)

    // 对 DStream 进行转换操作，将每一行数据做切分，形成一个个单词
    val wordStreams = inputDStream.flatMap(_.split(" "))

    // 将单词映射成元组（word,1）
    val wordPairsStreams = wordStreams.map((_, 1))

    // 使用window函数进行滑动窗口操作
    val reducedStream = wordPairsStreams.reduceByKeyAndWindow(
      (a, b) => a + b, // 增量计算函数
      (a, b) => a - b, // 反向增量计算函数,
      Seconds(10), Seconds(5))

    // 打印结果
    reducedStream.print()
  }
}
